package com.ecommerce.db

import com.ecommerce.utils.ConfigLoader
import com.typesafe.config.Config
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.log4j.Logger
import java.sql.{DriverManager, ResultSet}

object MySQLHandler {
  private val logger = Logger.getLogger(MySQLHandler.getClass)
  private val config: Config = ConfigLoader.loadConfig

  private val mysqlUrl: String = config.getString("mysql.url")
  private val mysqlUser: String = config.getString("mysql.user")
  private val mysqlPassword: String = config.getString("mysql.password")
  private val mysqlDriver: String = config.getString("mysql.driver")

  try {
    Class.forName(mysqlDriver)
    logger.info(s"成功加载MySQL驱动: $mysqlDriver")
  } catch {
    case e: ClassNotFoundException =>
      logger.error(s"找不到MySQL驱动类: $mysqlDriver", e)
      throw new RuntimeException("MySQL驱动加载失败", e)
  }

  def testConnection(): Boolean = {
    var connection: java.sql.Connection = null
    try {
      connection = DriverManager.getConnection(mysqlUrl, mysqlUser, mysqlPassword)
      logger.info("MySQL连接测试成功")
      true
    } catch {
      case e: Exception =>
        logger.error(s"MySQL连接测试失败: ${e.getMessage}", e)
        false
    } finally {
      if (connection != null) {
        try {
          connection.close()
        } catch {
          case e: Exception =>
            logger.warn("关闭测试连接时发生错误", e)
        }
      }
    }
  }

  def tableExists(tableName: String): Boolean = {
    var connection: java.sql.Connection = null
    var resultSet: ResultSet = null
    try {
      connection = DriverManager.getConnection(mysqlUrl, mysqlUser, mysqlPassword)
      val metaData = connection.getMetaData
      resultSet = metaData.getTables(null, null, tableName, Array("TABLE"))
      resultSet.next()
    } catch {
      case e: Exception =>
        logger.error(s"检查表[$tableName]是否存在时出错", e)
        false
    } finally {
      if (resultSet != null) resultSet.close()
      if (connection != null) connection.close()
    }
  }

  // 仅保留基础的表创建方法
  def createTable(connection: java.sql.Connection, createSql: String): Unit = {
    try {
      val statement = connection.createStatement()
      statement.execute(createSql)
    } catch {
      case e: Exception =>
        logger.error(s"执行创建表SQL失败: $createSql", e)
        throw e
    }
  }

  def readFromMySQL(spark: SparkSession, tableName: String): DataFrame = {
    if (!testConnection()) {
      throw new RuntimeException("MySQL连接不可用，无法读取数据")
    }

    spark.read
      .format("jdbc")
      .option("url", mysqlUrl)
      .option("dbtable", tableName)
      .option("user", mysqlUser)
      .option("password", mysqlPassword)
      .option("driver", mysqlDriver)
      .load()
  }

  def writeToMySQL(df: DataFrame, tableName: String, saveMode: String = "overwrite"): Unit = {
    if (!testConnection()) {
      throw new RuntimeException("MySQL连接不可用，无法写入数据")
    }

    df.write
      .format("jdbc")
      .option("url", mysqlUrl)
      .option("dbtable", tableName)
      .option("user", mysqlUser)
      .option("password", mysqlPassword)
      .option("driver", mysqlDriver)
      .mode(saveMode)
      .save()

    logger.info(s"成功将数据写入MySQL表: $tableName")
  }

  def readFromMySQLBySQL(spark: SparkSession, sqlQuery: String): DataFrame = {
    if (!testConnection()) {
      throw new RuntimeException("MySQL连接不可用，无法执行查询")
    }

    spark.read
      .format("jdbc")
      .option("url", mysqlUrl)
      .option("query", sqlQuery)
      .option("user", mysqlUser)
      .option("password", mysqlPassword)
      .option("driver", mysqlDriver)
      .load()
  }

  // 获取数据库连接的方法
  def getConnection(): java.sql.Connection = {
    DriverManager.getConnection(mysqlUrl, mysqlUser, mysqlPassword)
  }
}